package reactiveStream;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * @projectName: Webflux_demo
 * @package: reactiveStream
 * @className: ReactiveStreamDemo2
 * @author: 冷环渊 doomwatcher
 * @description: TODO
 * @date: 2021/12/14 0:25
 * @version: 1.0
 */
public class ReactiveStreamDemo2 {
    public static void main(String[] args) {
        // 1.创建一个 发布者
        SubmissionPublisher publisher = new SubmissionPublisher();
        // 2.创建一个 Processor
        ReactiveProcessor processor = new ReactiveProcessor();
        // 3 发布者将消息给processor来做处理之后转发到最终订阅者
        publisher.subscribe(processor);
        // 4.创建一个最终订阅者
        Flow.Subscriber subscriber = new Flow.Subscriber() {
            Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                System.out.println("创建订阅关系 ");
                subscription.request(1); //第一次需要发送一个 之后的都不需要了
            }

            @Override
            public void onNext(Object item) {
                System.out.println("接收数据：" + item);
                //接收数据 业务处理
                subscription.request(10);
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("发生错误了");
            }

            @Override
            public void onComplete() {
                System.out.println("数据发送完成了");
            }
        };
        processor.subscribe(subscriber);
        for (int i = 0; i < 100; i++) {
            System.out.println("发布数据" + i);
            // 4 发送数据
            publisher.submit("第" + i + "条hello reactive stream");
        }
        publisher.close();
        try {
            Thread.currentThread().join(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
